package cn.ac.iie.di.ban.ycl.cleaner.sender;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import org.apache.commons.configuration2.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-11-8 11:14:11
 * @version V1.0
 * @desc
 */
public class IvrMsgCountSenderThread implements Runnable {

    private static final Logger LOG = LoggerFactory.getLogger(IvrMsgCountSenderThread.class);
    MessageSender sender;
    long timeOut;
    long batchNum;
    List<Object[]> cacheDatas = new ArrayList<>();

    public IvrMsgCountSenderThread(Configuration conf) {
        String nameServer = conf.getString(Constants.RECEIVER_MQ_NAMESERVER);
        String topic = conf.getString(Constants.IVRMSG_COUNT_SEND_TOPIC);
        sender = new MessageSender(nameServer, topic,conf);
        timeOut = conf.getLong(Constants.SEND_QUEUE_TIME_OUT, 5000);
        batchNum = conf.getLong(Constants.SEND_QUEUE_BATCH_NUM, 10);
    }

    @Override
    public void run() {
        LOG.info("ivr count send thread start!");
        long startTime = System.currentTimeMillis();
        while (true) {
            try {
                REMessage message = (REMessage) CleanerMain.ivrMsgCountQueue.poll(1, TimeUnit.SECONDS);
                if (message == null) {
                    continue;
                }
                List<Object[]> datas = message.data;
                if (datas == null || datas.isEmpty()) {
                    continue;
                }
                cacheDatas.addAll(datas);
                long endTime = System.currentTimeMillis();
                long time = endTime - startTime;
                if (time < timeOut && cacheDatas.size() < batchNum) {
                    continue;
                }
                REMessage cacheMessage = new REMessage(message.colNames,
                        message.types, cacheDatas, message.docSchemaName,
                        message.userDesc);
                sender.sendMessage(cacheMessage);
                LOG.info("send ivr count messages num: {}", cacheDatas.size());
                cacheDatas.clear();
                startTime = System.currentTimeMillis();
            } catch (Exception ex) {
                LOG.error("send ivr count messages error!", ex);
            }
        }
    }

}
